#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
选股策略类
包含所有选股策略函数，只接收加工好的数据
"""

import pandas as pd
import numpy as np
from typing import List, Dict
from loguru import logger

class StockStrategies:
    """选股策略类"""
    
    @staticmethod
    def apply_strategy_to_all(data: pd.DataFrame, strategy_name: str = 'breakout_platform') -> pd.DataFrame:
        """对所有股票应用选股策略"""
        if not hasattr(StockStrategies, strategy_name):
            raise ValueError(f"无效的策略名称: {strategy_name}")

        grouped = data.groupby('security_id')
        results = []
        
        for security_id, group in grouped:
            group = group.sort_values('trade_date')
            
            try:
                if len(group) < 2:
                    continue
                strategy_func = getattr(StockStrategies, strategy_name)
                is_selected = strategy_func(group)
                
                if is_selected:
                    last_record = group.iloc[-1].copy()
                    last_record['strategy'] = strategy_name
                    results.append(last_record)
                    
            except Exception as e:
                logger.error(f"应用策略 {strategy_name} 到股票 {security_id} 时出错: {str(e)}")
                continue
        
        if results:
            return pd.DataFrame(results)
        else:
            return pd.DataFrame()

    @staticmethod
    def breakout_platform(data: pd.DataFrame, lookback_days: int = 15) -> bool:
        """突破平台策略"""
        if len(data) < lookback_days:
            return False
            
        ma_window = min(60, len(data))
        ma60 = data['close'].rolling(ma_window).mean().iloc[-1]
        
        recent_close = data['close'].iloc[-lookback_days:]
        if not all((recent_close - ma60).abs() / ma60 <= 0.05):
            return False
            
        avg_vol5 = data['volume'].rolling(5).mean().iloc[-1]
        last_bar = data.iloc[-1]
        if not (last_bar['volume'] > 1.5 * avg_vol5 and last_bar['close'] > last_bar['open']):
            return False
            
        for i in range(1, lookback_days + 1):
            close = data['close'].iloc[-i]
            deviation = (close - ma60) / ma60
            if -0.05 <= deviation <= 0.2:
                return True
                
        return False

    @staticmethod
    def airfield_strategy(data: pd.DataFrame, lookback_days: int = 15) -> bool:
        """停机坪策略"""
        min_days = lookback_days + 3
        if len(data) < min_days:
            return False
            
        vol_window = min(15, len(data))
        avg_vol15 = data['volume'].rolling(vol_window).mean()
        has_big_rise = False
        rise_day_idx = -1
        
        for i in range(1, lookback_days + 1):
            bar = data.iloc[-i]
            if bar['open'] <= 0:
                continue
                
            rise = (bar['close'] - bar['open']) / bar['open']
            if rise > 0.095 and bar['volume'] > 1.8 * avg_vol15.iloc[-i]:
                has_big_rise = True
                rise_day_idx = -i
                break
                
        if not has_big_rise:
            return False
            
        next_day1 = data.iloc[rise_day_idx + 1]
        if next_day1['open'] <= 0:
            return False
            
        rise1 = (next_day1['close'] - next_day1['open']) / next_day1['open']
        prev_close = data.iloc[rise_day_idx]['close']
        if not (next_day1['open'] > prev_close and 0 < rise1 < 0.03):
            return False
            
        for i in [2, 3]:
            next_day = data.iloc[rise_day_idx + i]
            prev_day = data.iloc[rise_day_idx + i - 1]
            
            if next_day['open'] <= 0 or prev_day['close'] <= 0:
                return False
                
            rise = (next_day['close'] - next_day['open']) / next_day['open']
            change = (next_day['close'] - prev_day['close']) / prev_day['close']
            
            if not (next_day['open'] > prev_day['close'] and 
                    0 < rise < 0.03 and 
                    abs(change) < 0.05):
                return False
                
        return True

    @staticmethod
    def volume_limit_down(data: pd.DataFrame) -> bool:
        """放量跌停策略"""
        if len(data) < 2:
            return False
            
        last_bar = data.iloc[-1]
        if last_bar['open'] <= 0:
            return False
            
        drop = (last_bar['close'] - last_bar['open']) / last_bar['open']
        if drop <= -0.095:
            avg_vol5 = data['volume'].rolling(5).mean().iloc[-1]
            if last_bar['volume'] > 1.5 * avg_vol5:
                return True
                
        return False

    @staticmethod
    def no_large_drawdown(data: pd.DataFrame, period: int = 60) -> bool:
        """
        无大幅回撤策略
        条件1: 当日收盘价比60日前的收盘价的涨幅小于0.6%
        条件2: 检查60日内是否有大幅回撤(单日跌幅>7%,高开低走>7%,两日累计跌幅>10%)
        """
        period = min(period, len(data) - 1)
        if len(data) < period + 1:
            return False
            
        price60 = data['close'].iloc[-period-1]
        rise = (data['close'].iloc[-1] - price60) / price60
        if rise >= 0.006:
            return False
            
        for i in range(1, period + 1):
            bar = data.iloc[-i]
            prev_bar = data.iloc[-i-1] if i < len(data)-1 else None
            
            if prev_bar is not None:
                daily_drop = (prev_bar['close'] - bar['close']) / prev_bar['close']
                if daily_drop >= 0.07:
                    return False
                    
            if bar['open'] > 0:
                open_close_drop = (bar['open'] - bar['close']) / bar['open']
                if open_close_drop >= 0.07:
                    return False
                    
            if i > 1 and prev_bar is not None:
                two_day_drop = (data.iloc[-i-1]['close'] - bar['close']) / data.iloc[-i-1]['close']
                if two_day_drop >= 0.1:
                    return False
                    
        return True

    # 其他策略函数占位符
    @staticmethod
    def flag_pattern(data: pd.DataFrame, min_trading_days: int = 60) -> bool:
        """高而窄的旗形策略"""
        min_trading_days = min(min_trading_days, len(data))
        if len(data) < min_trading_days:
            return False
            
        lookback_start = 24
        lookback_end = 10

        if len(data) < lookback_start:
            return False
            
        min_price = data['close'].iloc[-lookback_start:-lookback_end].min()
        
        if data['close'].iloc[-1] / min_price < 1.9:
            return False
            
        has_two_big_rises = False
        for i in range(lookback_start, lookback_end + 1, -1):
            if i < 2:
                continue
                
            bar1 = data.iloc[-i]
            bar2 = data.iloc[-i+1]
            
            if bar1['open'] <= 0 or bar2['open'] <= 0:
                continue
                
            rise1 = (bar1['close'] - bar1['open']) / bar1['open']
            rise2 = (bar2['close'] - bar2['open']) / bar2['open']
            
            if rise1 >= 0.095 and rise2 >= 0.095:
                has_two_big_rises = True
                break
                
        return has_two_big_rises
    @staticmethod
    def four_up_three_down(data: pd.DataFrame, min_trading_days: int = 60) -> bool:
        """四连板后三连跌策略"""
        min_trading_days = min(min_trading_days, len(data))
        if len(data) < min_trading_days:
            return False

        if min_trading_days < 30:
            return False

        up_streak = 0
        up_start_idx = -1
        up_start_price = 0
        
        for i in range(1, min_trading_days + 1):
            bar = data.iloc[-i]
            if bar['close'] == bar['upper_px']:
                up_streak += 1
                if up_streak == 1:
                    up_start_idx = -i
                    up_start_price = bar['close']
            else:
                if up_streak >= 4:
                    break
                up_streak = 0
                
        if up_streak < 4:
            return False
            
        down_streak = 0
        for i in range(up_start_idx + 1, 0):
            bar = data.iloc[i]
            if bar['close'] == bar['lower_px']:
                down_streak += 1
                if down_streak >= 2:
                    break
            else:
                down_streak = 0
                
        if down_streak < 2:
            return False
            
        current_price = data.iloc[-1]['close']
        price_diff = abs(current_price - up_start_price) / up_start_price
        if price_diff > 0.05:
            return False
            
        return True

    @staticmethod
    def near_high_with_limited_ups(data: pd.DataFrame, lookback_days: int = 20) -> bool:
        """
        接近高点且有2-3个涨停策略
        条件1: 20日内无连续涨停
        条件2: 20日内有2-3个涨停
        条件3: 当前价格接近20日高点(±3%)
        """
        if len(data) < lookback_days:
            return False
            
        consecutive_ups = 0
        for i in range(1, lookback_days + 1):
            bar = data.iloc[-i]
            if bar['close'] == bar['upper_px']:
                consecutive_ups += 1
                if consecutive_ups >= 2:
                    return False
            else:
                consecutive_ups = 0
                
        up_count = 0
        for i in range(1, lookback_days + 1):
            bar = data.iloc[-i]
            if bar['close'] == bar['upper_px']:
                up_count += 1
                
        if up_count < 2 or up_count > 3:
            return False
            
        recent_high = data['high'].iloc[-lookback_days:].max()
        current_price = data.iloc[-1]['close']
        price_diff = abs(current_price - recent_high) / recent_high
        
        if price_diff > 0.03:
            return False
            
        return True

    @staticmethod
    def stable_trend_with_volume(data: pd.DataFrame, lookback_days: int = 20) -> bool:
        """
        趋势稳定且最近放量/波动策略
        条件1: 20日内趋势稳定性高(R方值高)
        条件2: 20日内波动率低
        条件3: 最近2天有放量或波动率放大
        """
        if len(data) < lookback_days:
            return False
            
        recent_data = data.iloc[-lookback_days:]
        
        x = np.arange(len(recent_data))
        y = recent_data['close'].values.astype(float)
        slope, intercept = np.polyfit(x, y, 1)
        y_pred = slope * x + intercept
        r_squared = 1 - np.sum((y - y_pred) ** 2) / np.sum((y - np.mean(y)) ** 2)
        
        volatility = recent_data['close'].pct_change().std()
        
        if len(recent_data) <= 2:
            return False
            
        recent_2d = recent_data.iloc[-2:]
        avg_volume = recent_data['volume'].iloc[:-2].mean()
        recent_volume = recent_2d['volume'].mean()
        volume_increase = recent_volume / avg_volume if avg_volume > 0 else 0
        
        recent_volatility = recent_2d['close'].pct_change().std()
        volatility_increase = recent_volatility / volatility if volatility > 0 else 0
        
        return (r_squared > 0.7 and 
                volatility < 0.03 and 
                (volume_increase > 1.5 or volatility_increase > 1.5))
    @staticmethod
    def double_bottom(data: pd.DataFrame, min_days: int = 40) -> bool:
        """双底(W底)突破策略"""
        if len(data) < min_days:
            return False
            
        lows = data['low'].rolling(5,min_periods=1).min()
        min1_idx = lows.idxmin()
        min1 = lows[min1_idx]

        min1_pos = data.index.get_loc(min1_idx)
        if min1_pos + 5 >= len(data):
            return False

        remaining_data = data.iloc[min1_pos + 5:]
        if len(remaining_data) < 5:
            return False
            
        min2_idx = remaining_data['low'].idxmin()
        min2 = data.loc[min2_idx, 'low']
        
        if min1 != min2 and abs(min1 - min2) / min(min1, min2) > 0.03:
            return False
            
        vol1 = data.loc[min1_idx, 'volume']
        vol2 = data.loc[min2_idx, 'volume']
        if vol2 > vol1 * 0.7:
            return False

        neck_line = data.loc[min1_idx : min2_idx, 'high'].max()
        
        last_bar = data.iloc[-1]
        avg_vol5 = data['volume'].rolling(5).mean().iloc[-1]
        return (last_bar['close'] > neck_line and 
                last_bar['volume'] > 1.5 * avg_vol5)
    @staticmethod
    def head_shoulder_bottom(data: pd.DataFrame, min_days: int = 60) -> bool:
        """头肩底突破策略"""
        if len(data) < min_days:
            return False
            
        head_idx = data['low'].idxmin()
        head = data.loc[head_idx, 'low']
        
        left_data = data.loc[:head_idx]
        left_shoulder_idx = left_data['high'].idxmax()
        
        right_data = data.loc[head_idx:]
        right_shoulder_idx = right_data['high'].idxmax()
        
        if (left_shoulder_idx >= head_idx or 
            right_shoulder_idx <= head_idx):
            return False
            
        right_vol = data.loc[right_shoulder_idx, 'volume']
        left_vol = data.loc[left_shoulder_idx, 'volume']
        if right_vol <= left_vol:
            return False
            
        neck_line = max(data.loc[left_shoulder_idx, 'high'], 
                       data.loc[right_shoulder_idx, 'high'])
        
        last_bar = data.iloc[-1]
        avg_vol5 = data['volume'].rolling(5).mean().iloc[-1]
        return (last_bar['close'] > neck_line and 
                last_bar['volume'] > 2 * avg_vol5)
    @staticmethod
    def rounding_bottom(data: pd.DataFrame, min_days: int = 50) -> bool:
        """圆弧底突破策略"""
        if len(data) < min_days:
            return False
            
        price_changes = data['close'].pct_change().std()
        if price_changes > 0.03:
            return False
            
        last_bar = data.iloc[-1]
        avg_vol10 = data['volume'].rolling(10).mean().iloc[-1]
        if last_bar['volume'] <= 1.8 * avg_vol10:
            return False
            
        exp12 = data['close'].ewm(span=12, adjust=False).mean()
        exp26 = data['close'].ewm(span=26, adjust=False).mean()
        macd = exp12 - exp26
        signal = macd.ewm(span=9, adjust=False).mean()
        
        return macd.iloc[-1] > signal.iloc[-1] and macd.iloc[-2] <= signal.iloc[-2]
    @staticmethod
    def morning_star(data: pd.DataFrame, lookback_days: int = 10) -> bool:
        """早晨之星策略"""
        if len(data) < 3:
            return False
            
        day1 = data.iloc[-3]
        day2 = data.iloc[-2] 
        day3 = data.iloc[-1]
        
        if not (day1['close'] < day1['open'] and
                abs(day2['close'] - day2['open']) / day2['open'] < 0.01 and
                day3['close'] > day3['open']):
            return False
            
        day1_drop = (day1['open'] - day1['close']) / day1['open']
        day3_rise = (day3['close'] - day3['open']) / day3['open']
        if day3_rise < day1_drop * 0.5:
            return False
            
        avg_vol = data['volume'].iloc[:-3].mean()
        return day3['volume'] > 1.5 * avg_vol
    @staticmethod
    def v_shape_bottom(data: pd.DataFrame, lookback_days: int = 20) -> bool:
        """V形底(单针探底)策略"""
        if len(data) < lookback_days:
            return False
            
        has_long_lower_shadow = False
        shadow_day_idx = -1
        
        for i in range(1, lookback_days + 1):
            bar = data.iloc[-i]
            if bar['open'] <= 0 or bar['close'] <= 0:
                continue
                
            body = abs(bar['close'] - bar['open'])
            lower_shadow = bar['open'] - bar['low'] if bar['close'] > bar['open'] else bar['close'] - bar['low']
            
            if lower_shadow > body * 2:
                has_long_lower_shadow = True
                shadow_day_idx = -i
                break
                
        if not has_long_lower_shadow:
            return False
            
        if len(data) < abs(shadow_day_idx) + 3:
            return False
            
        for i in range(1, 4):
            prev_bar = data.iloc[shadow_day_idx + i - 1]
            curr_bar = data.iloc[shadow_day_idx + i]
            if curr_bar['close'] <= prev_bar['close']:
                return False
                
        exp12 = data['close'].ewm(span=12, adjust=False).mean()
        exp26 = data['close'].ewm(span=26, adjust=False).mean()
        macd = exp12 - exp26
        
        price_lows = data['low'].rolling(5).min()
        macd_highs = macd.rolling(5).max()
        
        return (price_lows.iloc[-1] < price_lows.iloc[-5] and 
                macd.iloc[-1] > macd.iloc[-5])
    @staticmethod
    def latent_bottom(data: pd.DataFrame, min_days: int = 60) -> bool:
        """潜伏底突破策略"""
        if len(data) < min_days:
            return False
            
        price_volatility = data['close'].pct_change().std()
        if price_volatility >= 0.03:
            return False
            
        avg_vol20 = data['volume'].rolling(20).mean()
        hist_avg_vol20 = avg_vol20.mean()
        if avg_vol20.iloc[-1] >= hist_avg_vol20 * 0.5:
            return False
            
        last_bar = data.iloc[-1]
        if last_bar['volume'] <= 3 * avg_vol20.iloc[-1]:
            return False
            
        recent_high = data['high'].rolling(20).max().iloc[-1]
        recent_low = data['low'].rolling(20).min().iloc[-1]
        return last_bar['close'] > recent_high
    @staticmethod
    def rising_flag(data: pd.DataFrame, lookback_days: int = 20) -> bool:
        """上升旗形突破策略"""
        if len(data) < max(lookback_days, 5):
            return False
            
        try:
            price10 = data['close'].iloc[-lookback_days]
            rise = (data['close'].iloc[-1] - price10) / price10
            if rise <= 0.15:
                return False
        except Exception as e:
            # Error1 in rising_flag: {e}
            return False
            
        try:
            highs = data['high'].iloc[-5:]
            if len(highs) < 2:
                return False
            for i in range(len(highs)-1):
                if highs.iloc[i] <= highs.iloc[i+1]:
                    return False    
        except Exception as e:
            # Error2 in rising_flag: {e}
            return False
            
        try:
            flag_high = highs.max()
            last_bar = data.iloc[-1]
            avg_vol10 = data['volume'].rolling(10).mean().iloc[-1]
            return (last_bar['close'] > flag_high and 
                    last_bar['volume'] > 2 * avg_vol10)
        except Exception as e:
            # Error3 in rising_flag: {e}
            return False
    @staticmethod
    def morning_sun(data: pd.DataFrame) -> bool:
        """旭日东升策略"""
        if len(data) < 2:
            return False
            
        prev_day = data.iloc[-2]
        curr_day = data.iloc[-1]
        
        if prev_day['close'] >= prev_day['open']:
            return False
            
        if not (curr_day['open'] > prev_day['close'] and 
                curr_day['close'] > curr_day['open']):
            return False
            
        if curr_day['close'] <= prev_day['open']:
            return False
            
        avg_vol5 = data['volume'].rolling(5).mean().iloc[-1]
        return curr_day['volume'] > 1.5 * avg_vol5
    @staticmethod
    def cup_with_handle(data: pd.DataFrame, min_days: int = 60) -> bool:
        """杯柄形态策略"""
        if len(data) < min_days:
            return False
            
        cup_bottom = data['low'].iloc[-min_days:].min()
        cup_bottom_idx = data['low'].iloc[-min_days:].idxmin()
        cup_bottom_pos = data.index.get_loc(cup_bottom_idx)
        
        if cup_bottom_pos + 5 >= len(data):
            return False

        left_high = data['high'].iloc[-min_days:cup_bottom_pos].max()
        right_high = data['high'].iloc[cup_bottom_pos:].max()
        
        if abs(right_high - left_high) / left_high > 0.1:
            return False
            
        handle_start_idx = data['high'].iloc[cup_bottom_pos:].idxmax()
        handle_data = data.loc[handle_start_idx:]
        
        if len(handle_data) < 5:
            return False
            
        handle_low = handle_data['low'].min()
        if (right_high - handle_low) / right_high > 0.08:
            return False
            
        last_bar = data.iloc[-1]
        avg_vol10 = data['volume'].rolling(10).mean().iloc[-1]
        return (last_bar['close'] > right_high and 
                last_bar['volume'] > 1.5 * avg_vol10)
    @staticmethod
    def triangle_breakout(data: pd.DataFrame, min_days: int = 30) -> bool:
        """三角形突破策略"""
        if len(data) < min_days:
            return False
            
        highs = data['high'].iloc[-min_days:]
        lows = data['low'].iloc[-min_days:]
        
        x = np.arange(len(highs))
        high_slope = np.polyfit(x, highs, 1)[0]
        low_slope = np.polyfit(x, lows, 1)[0]
        
        if not (high_slope < -0.001 and low_slope > 0.001):
            return False
            
        resistance = highs.max()
        last_bar = data.iloc[-1]
        avg_vol10 = data['volume'].rolling(10).mean().iloc[-1]
        
        if not (last_bar['close'] > resistance and 
                last_bar['volume'] > 1.8 * avg_vol10):
            return False
            
        exp12 = data['close'].ewm(span=12, adjust=False).mean()
        exp26 = data['close'].ewm(span=26, adjust=False).mean()
        macd = exp12 - exp26
        signal = macd.ewm(span=9, adjust=False).mean()
        
        return (macd.iloc[-1] > signal.iloc[-1] or 
                abs(macd.iloc[-1] - signal.iloc[-1]) / signal.iloc[-1] < 0.02)
    @staticmethod
    def volume_breakout(data: pd.DataFrame, min_days: int = 20) -> bool:
        """放量突破策略"""
        if len(data) < min_days:
            return False
            
        volatility = data['close'].pct_change().rolling(5).std()
        if not all(volatility.iloc[-5:-1] > volatility.iloc[-1]):
            return False
            
        avg_vol20 = data['volume'].rolling(min_days).mean().iloc[-1]
        last_bar = data.iloc[-1]
        if last_bar['volume'] <= 3 * avg_vol20:
            return False
            
        return last_bar['close'] == data['high'].iloc[-min_days:].max()
    @staticmethod
    def rectangle_breakout(data: pd.DataFrame, min_days: int = 40) -> bool:
        """矩形整理突破策略"""
        if len(data) < min_days:
            return False
            
        highs = data['high'].iloc[-min_days:]
        lows = data['low'].iloc[-min_days:]
        
        price_range = highs.max() - lows.min()
        if price_range / lows.min() > 0.08:
            return False
            
        resistance = highs.max()
        last_bar = data.iloc[-1]
        avg_vol20 = data['volume'].rolling(20).mean().iloc[-1]
        
        if not (last_bar['close'] > resistance and 
                last_bar['volume'] > 2 * avg_vol20):
            return False
            
        delta = data['close'].diff()
        gain = delta.where(delta > 0, 0)
        loss = -delta.where(delta < 0, 0)
        avg_gain = gain.rolling(14).mean()
        avg_loss = loss.rolling(14).mean()
        rs = avg_gain / avg_loss
        rsi = 100 - (100 / (1 + rs))
        
        return rsi.iloc[-1] > 50
    @staticmethod
    def low_breakout_strategy(data: pd.DataFrame, lookback_days: int = 20) -> bool:
        """低位突破策略"""
        if len(data) < lookback_days + 1:
            return False
            
        prev_data = data.iloc[-lookback_days-1:-1]
        
        high_20 = prev_data['high'].max()
        low_20 = prev_data['low'].min()
        price_range = (high_20 - low_20) / low_20
        if price_range >= 0.15:
            return False
            
        vol_mean = prev_data['volume'].mean()
        vol_std = prev_data['volume'].std()
        if vol_mean == 0 or vol_std / vol_mean >= 0.3:
            return False
            
        last_day = data.iloc[-1]
        if last_day['close'] == last_day['upper_px'] and last_day['volume'] >= 3 * vol_mean:
            return True
            
        return False
    @staticmethod
    def high_breakout_strategy(data: pd.DataFrame, lookback_days: int = 20) -> bool:
        """高位突破选股策略"""
        if len(data) < lookback_days + 2:
            return False
            
        prev_20d_high = data['high'].iloc[-lookback_days-2:-2].max()
        current_close = data['close'].iloc[-1]
        if current_close <= prev_20d_high:
            return False
        
        prev_day = data.iloc[-2]
        today = data.iloc[-1]
        
        is_prev_day_limit_up = prev_day['close'] == prev_day['upper_px']
        is_today_down = today['close'] < today['open'] and today['open'] < prev_day['close']
        
        return is_prev_day_limit_up and is_today_down
    @staticmethod
    def low_volume_rise_with_stable_baseline(data: pd.DataFrame, lookback_days: int = 20) -> bool:
        """缩量拉涨策略(前期波动平稳)"""
        if len(data) < lookback_days:
            return False
            
        recent_5d = data.iloc[-5:]
        if len(recent_5d) < 5:
            return False
            
        price_rise = (recent_5d['close'].iloc[-1] - recent_5d['close'].iloc[0]) / recent_5d['close'].iloc[0]
        if price_rise <= 0:
            return False
            
        avg_vol5 = data['volume'].rolling(5).mean().iloc[-6]
        recent_vol = recent_5d['volume'].mean()
        if recent_vol >= avg_vol5 * 0.9:
            return False
            
        baseline_data = data.iloc[-lookback_days:-5]
        price_std = baseline_data['close'].std()
        price_mean = baseline_data['close'].mean()
        if price_std / price_mean >= 0.15:
            return False
            
        x = np.arange(len(baseline_data))
        y = baseline_data['close'].values.astype(float)
        slope, intercept = np.polyfit(x, y, 1)
        y_pred = slope * x + intercept
        r_squared = 1 - np.sum((y - y_pred) ** 2) / np.sum((y - np.mean(y)) ** 2)
        
        return r_squared > 0.7
    @staticmethod
    def ma_crossover(data: pd.DataFrame, short_period: int = 5, long_period: int = 20) -> bool:
        """均线金叉策略(5日线上穿20日线)"""
        if len(data) < long_period + 1:
            return False
            
        ma5 = data['close'].rolling(short_period).mean()
        ma20 = data['close'].rolling(long_period).mean()
        
        if ma5.iloc[-1] <= ma20.iloc[-1]:
            return False
            
        if ma5.iloc[-2] > ma20.iloc[-2]:
            return False
            
        avg_vol5 = data['volume'].rolling(5).mean().iloc[-1]
        if data['volume'].iloc[-1] <= avg_vol5:
            return False
            
        return True
    @staticmethod
    def limit_up_then_down(data: pd.DataFrame, lookback_days: int = 10) -> bool:
        """涨停后收阴策略"""
        if len(data) < lookback_days + 1:
            return False
            
        for i in range(1, lookback_days + 1):
            if data.iloc[-i]['high'] == data.iloc[-i]['upper_px']:
                if len(data) <= -i + 1:
                    return False

                next_day = data.iloc[-i + 1]
                if next_day['close'] >= next_day['open']:
                    return False
                    
                premium = (next_day['high'] - next_day['pre_close']) / next_day['pre_close']
                if premium <= 0.01:
                    return True
                else:
                    return False

        return False
    @staticmethod
    def rising_close_5days(data: pd.DataFrame, lookback_days: int = 5) -> bool:
        """判断股票是否连续5天收盘价递增"""
        if len(data) < lookback_days:
            return False
        
        closes = data['close'].tail(lookback_days).values
        
        first_day_close = closes[0]
        last_day_close = closes[-1]
        zf = (last_day_close - first_day_close) / first_day_close
        if zf < 0.05:
            return False

        for i in range(1, lookback_days):
            if closes[i] <= closes[i-1]:
                return False
        
        return True
    @staticmethod
    def lowest_in_60days(data: pd.DataFrame) -> bool:
        """最近60日最低位策略"""
        if len(data) < 60:
            return False
        recent_data = data.tail(60)
        current_close = recent_data.iloc[-1]['close']
        lowest_close = recent_data['close'].min()
        
        highest_high = recent_data['high'].max()
        lowest_low = recent_data['low'].min()
        
        amplitude = (highest_high - lowest_low) / lowest_low
        
        return current_close == lowest_close and amplitude > 0.3
    @staticmethod
    def three_yin_not_break_yang(data: pd.DataFrame, lookback_days: int = 20) -> bool:
        """三阴不破阳策略"""
        if len(data) < lookback_days + 3:
            return False
            
        limit_up_idx = -1
        limit_up_day = None
        
        for i in range(1, lookback_days + 1):
            if data.iloc[-i]['close'] == data.iloc[-i]['upper_px']:
                limit_up_idx = -i
                limit_up_day = data.iloc[-i]
                break
                
        if limit_up_idx == -1 or limit_up_day is None:
            return False
            
        if len(data) < abs(limit_up_idx) + 3:
            return False
            
        limit_up_volume = limit_up_day['volume']
        limit_up_low = limit_up_day['low']
        limit_up_close = limit_up_day['close']
        
        if abs(limit_up_idx) >= 6:
            avg_vol_5 = data['volume'].iloc[limit_up_idx-5:limit_up_idx].mean()
            if limit_up_volume < avg_vol_5 * 1.5:
                return False
        
        three_yin_days = []
        for i in range(1, 4):
            day_idx = limit_up_idx + i
            if day_idx >= 0:
                return False
                
            day_data = data.iloc[day_idx]
            
            if day_data['close'] >= day_data['open']:
                return False
                
            if day_data['volume'] >= limit_up_volume * 0.8:
                return False
                
            if day_data['low'] < limit_up_low:
                return False
                
            day_drop = (day_data['open'] - day_data['close']) / day_data['open']
            if day_drop > 0.05:
                return False
                
            three_yin_days.append(day_data)
            
        if len(three_yin_days) != 3:
            return False
            
        total_drop = (three_yin_days[0]['open'] - three_yin_days[2]['close']) / three_yin_days[0]['open']
        limit_up_rise = (limit_up_close - limit_up_day['open']) / limit_up_day['open']
        if total_drop > limit_up_rise * 0.5:
            return False
            
        if len(data) >= abs(limit_up_idx) + 4:
            day4_idx = limit_up_idx + 4
            if day4_idx < 0:
                day4_data = data.iloc[day4_idx]
                if day4_data['close'] > three_yin_days[2]['close']:
                    return True
                    
        return True
    @staticmethod
    def limit_up_rubbing_line(data: pd.DataFrame, lookback_days: int = 20) -> bool:
        """涨停揉搓线策略"""
        if len(data) < lookback_days + 3:
            return False
            
        limit_up_idx = -1
        limit_up_day = None
        
        for i in range(1, lookback_days + 1):
            if data.iloc[-i]['close'] == data.iloc[-i]['upper_px']:
                limit_up_idx = -i
                limit_up_day = data.iloc[-i]
                break
                
        if limit_up_idx == -1 or limit_up_day is None:
            return False
            
        if len(data) < abs(limit_up_idx) + 2:
            return False
            
        limit_up_volume = limit_up_day['volume']
        limit_up_close = limit_up_day['close']
        
        if abs(limit_up_idx) >= 6:
            avg_vol_5 = data['volume'].iloc[limit_up_idx-5:limit_up_idx].mean()
            if limit_up_volume < avg_vol_5 * 1.3:
                return False
        
        day1_idx = limit_up_idx + 1
        if day1_idx >= 0:
            return False
            
        day1_data = data.iloc[day1_idx]
        
        if day1_data['close'] >= day1_data['open']:
            return False
            
        body_length = abs(day1_data['close'] - day1_data['open'])
        upper_shadow = day1_data['high'] - max(day1_data['open'], day1_data['close'])
        if upper_shadow <= body_length * 1.1:
            return False
            
        if day1_data['volume'] >= limit_up_volume * 0.7:
            return False
            
        if day1_data['high'] <= limit_up_close:
            return False
            
        day2_idx = limit_up_idx + 2
        if day2_idx >= 0:
            return False
            
        day2_data = data.iloc[day2_idx]
        
        if day2_data['close'] <= day2_data['open']:
            return False
            
        body_length2 = abs(day2_data['close'] - day2_data['open'])
        lower_shadow = min(day2_data['open'], day2_data['close']) - day2_data['low']
        if lower_shadow <= body_length2 * 1.1:
            return False
            
        if day2_data['volume'] >= limit_up_volume * 0.7:
            return False
            
        if day2_data['close'] <= day1_data['close']:
            return False
            
        if day2_data['close'] > limit_up_close * 1.05:
            return False
            
        return True
    @staticmethod
    def limit_up_shrinking_volume(data: pd.DataFrame, lookback_days: int = 20) -> bool:
        """涨停缩倍量策略"""
        if len(data) < lookback_days + 4:
            return False
            
        limit_up_idx = -1
        limit_up_day = None
        
        for i in range(1, lookback_days + 1):
            if data.iloc[-i]['close'] == data.iloc[-i]['upper_px']:
                limit_up_idx = -i
                limit_up_day = data.iloc[-i]
                break
                
        if limit_up_idx == -1 or limit_up_day is None:
            return False
            
        if len(data) < abs(limit_up_idx) + 4:
            return False
            
        limit_up_volume = limit_up_day['volume']
        limit_up_close = limit_up_day['close']
        
        day1_idx = limit_up_idx + 1
        if day1_idx >= 0:
            return False
            
        day1_data = data.iloc[day1_idx]
        
        if day1_data['close'] >= day1_data['open']:
            return False
            
        if day1_data['volume'] >= limit_up_volume * 0.5:
            return False
                    
        return True
    @staticmethod
    def calculate_kdj(data: pd.DataFrame, n: int = 9, m1: int = 3, m2: int = 3) -> tuple:
        """计算KDJ指标"""
        high = data['high']
        low = data['low']
        close = data['close']
        
        low_n = low.rolling(n).min()
        high_n = high.rolling(n).max()
        rsv = (close - low_n) / (high_n - low_n) * 100
        
        k = pd.Series(index=data.index, dtype=float)
        k.iloc[0] = 50.0
        for i in range(1, len(data)):
            k.iloc[i] = (m1 - 1) / m1 * k.iloc[i-1] + 1 / m1 * rsv.iloc[i]
        
        d = pd.Series(index=data.index, dtype=float)
        d.iloc[0] = 50.0
        for i in range(1, len(data)):
            d.iloc[i] = (m2 - 1) / m2 * d.iloc[i-1] + 1 / m2 * k.iloc[i]
        
        j = 3 * k - 2 * d
        
        return k, d, j

    @staticmethod
    def calculate_rsi(data: pd.DataFrame, period: int = 14) -> pd.Series:
        """计算RSI指标"""
        delta = data['close'].diff()
        gain = (delta.where(delta > 0, 0)).rolling(window=period).mean()
        loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean()
        rs = gain / loss
        rsi = 100 - (100 / (1 + rs))
        return rsi

    @staticmethod
    def calculate_max_drawdown(data: pd.DataFrame, lookback_days: int = 20) -> float:
        """计算最大回撤"""
        if len(data) < lookback_days:
            return 0.0
        
        recent_data = data.tail(lookback_days)
        high_water_mark = recent_data['close'].expanding().max()
        drawdown = (recent_data['close'] - high_water_mark) / high_water_mark * 100
        return abs(drawdown.min())

    @staticmethod
    def oversold_rebound_technical_turning_point(data: pd.DataFrame, lookback_days: int = 20) -> bool:
        """超跌反弹+技术拐点策略"""
        if len(data) < max(lookback_days, 30):
            return False
        
        max_drawdown = StockStrategies.calculate_max_drawdown(data, lookback_days)
        if max_drawdown <= 15.0:
            return False
        
        k, d, j = StockStrategies.calculate_kdj(data)
        rsi = StockStrategies.calculate_rsi(data)
        ma5 = data['close'].rolling(5).mean()
        
        if len(k) < 2 or len(rsi) < 2 or len(ma5) < 1:
            return False
        
        today_k = k.iloc[-1]
        today_d = d.iloc[-1]
        yesterday_k = k.iloc[-2]
        yesterday_d = d.iloc[-2]
        
        kdj_golden_cross = (today_k > today_d and yesterday_k <= yesterday_d)
        
        today_rsi = rsi.iloc[-1]
        yesterday_rsi = rsi.iloc[-2]
        
        rsi_rebound = (today_rsi > 30 and today_rsi > yesterday_rsi)
        
        today_close = data['close'].iloc[-1]
        today_ma5 = ma5.iloc[-1]
        
        ma5_recovery = (today_close > today_ma5)
        
        return kdj_golden_cross or rsi_rebound or ma5_recovery
    @staticmethod
    def positive_net_inflow_3days(data: pd.DataFrame, days: int = 3) -> bool:
        """
        最近3日净流入为正的股票策略
        条件: 最近3个交易日的netamt字段都为正数
        """
        if len(data) < days:
            return False
            
        if 'netamt' not in data.columns:
            logger.warning("数据中缺少netamt字段")
            return False
            
        recent_netamt = data['netamt'].tail(days)
        
        if len(recent_netamt) < days:
            return False
            
        netamt_positive = all(recent_netamt > 0)
        
        if netamt_positive:
            logger.info(f"股票满足最近{days}日净流入都为正的条件")
            logger.info(f"  净流入额: {recent_netamt.values}")
            
        return netamt_positive
    @staticmethod
    def positive_big_net_inflow_3days(data: pd.DataFrame, days: int = 3) -> bool:
        """
        最近3日主力净流入为正的股票策略
        条件: 最近3个交易日的bignetamt字段都为正数
        """
        if len(data) < days:
            return False
            
        if 'bignetamt' not in data.columns:
            logger.warning("数据中缺少bignetamt字段")
            return False
            
        recent_bignetamt = data['bignetamt'].tail(days)
        
        if len(recent_bignetamt) < days:
            return False
            
        bignetamt_positive = all(recent_bignetamt > 0)
        
        if bignetamt_positive:
            logger.info(f"股票满足最近{days}日主力净流入都为正的条件")
            logger.info(f"  主力净流入额: {recent_bignetamt.values}")
            
        return bignetamt_positive
    @staticmethod
    def positive_net_and_big_net_inflow_3days(data: pd.DataFrame, days: int = 3) -> bool:
        """最近3日净流入和主力净流入都为正的股票策略"""
        if len(data) < days:
            return False
            
        if 'netamt' not in data.columns:
            logger.warning("数据中缺少netamt字段")
            return False
            
        if 'bignetamt' not in data.columns:
            logger.warning("数据中缺少bignetamt字段")
            return False
            
        recent_netamt = data['netamt'].tail(days)
        recent_bignetamt = data['bignetamt'].tail(days)
        
        if len(recent_netamt) < days or len(recent_bignetamt) < days:
            return False
            
        netamt_positive = all(recent_netamt > 0)
        bignetamt_positive = all(recent_bignetamt > 0)
        
        both_positive = netamt_positive and bignetamt_positive
        
        if both_positive:
            logger.info(f"股票满足最近{days}日净流入和主力净流入都为正的条件")
            logger.info(f"  净流入额: {recent_netamt.values}")
            logger.info(f"  主力净流入额: {recent_bignetamt.values}")
            
        return both_positive
    @staticmethod
    def limit_up_pullback_restart_strategy(data: pd.DataFrame, lookback_days: int = 100) -> bool:
        """涨停回调再启动策略 (通达信公式转换)"""
        if len(data) < lookback_days:
            return False
            
        recent_60d = data.iloc[-60:]
        has_limit_up_60d = False
        for i in range(len(recent_60d)):
            if i == 0:
                continue
            current_close = recent_60d.iloc[i]['close']
            prev_close = recent_60d.iloc[i-1]['close']
            if prev_close > 0 and current_close >= prev_close * 1.098:
                has_limit_up_60d = True
                break
                
        if not has_limit_up_60d:
            return False
            
        recent_100d = data.iloc[-lookback_days:]
        limit_up_count = 0
        for i in range(len(recent_100d)):
            if i == 0:
                continue
            current_close = recent_100d.iloc[i]['close']
            prev_close = recent_100d.iloc[i-1]['close']
            if prev_close > 0 and current_close >= prev_close * 1.098:
                limit_up_count += 1
                
        if limit_up_count < 2:
            return False
            
        current_high = data.iloc[-1]['high']
        prev_close = data.iloc[-2]['close']
        if prev_close <= 0:
            return False
        high_rise = (current_high - prev_close) / prev_close
        if high_rise <= 0.0007:
            return False
            
        last_limit_up_days = float('inf')
        for i in range(1, min(21, len(data))):
            if i >= len(data):
                break
            current_close = data.iloc[-i]['close']
            prev_close = data.iloc[-i-1]['close'] if i+1 < len(data) else 0
            if prev_close > 0 and current_close >= prev_close * 1.098:
                last_limit_up_days = i
                break
                
        if last_limit_up_days >= 20:
            return False
            
        if len(data) < 2:
            return False
        recent_2d_high = data['high'].iloc[-2:].max()
        if current_high < recent_2d_high:
            return False
            
        security_id = str(data.iloc[-1]['security_id'])
        if (security_id.startswith('300') or 
            security_id.startswith('301') or 
            security_id.startswith('688') or 
            security_id.startswith('8')):
            return False
            
        current_close = data.iloc[-1]['close']
        if current_close <= prev_close:
            return False
            
        if len(data) < 3:
            return False
        two_days_ago_close = data.iloc[-3]['close']
        if two_days_ago_close <= 0:
            return False
        two_day_rise = (current_close - two_days_ago_close) / two_days_ago_close
        if two_day_rise >= 0.15:
            return False
            
        return True
